package window

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, SlidingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time


object WindowFunctionPractise {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream1: DataStream[Click] = env.socketTextStream("localhost", 9999)
      .map(line => {
        val strs = line.split(",")
        Click(strs(0).toInt, strs(1), strs(2).toLong, strs(3).toInt)
      })
    //
    //    stream1.countWindowAll(5,2)
    //      .sum("age")
    //      .print()

//
//        stream1.windowAll(GlobalWindows.create())
//            .trigger(CountTrigger.of(2))
//            .evictor(CountEvictor.of(5))
//            .sum("age")
//            .pr

    //    stream1.windowAll(GlobalWindows.create())
    //        .trigger(new MyCountTrigger1())
    //        .evictor(new MyEvictor1())
    //        .sum("age")
    //        .print()
    //

    //    stream1.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    //      .sum("age")
    //      .print




    stream1
      .windowAll(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(2)))
      //.sum("age")
      .apply(new MyWindowFunction)
      .print()

    stream1.windowAll(SlidingEventTimeWindows.of(Time.milliseconds(5000), Time.milliseconds(2000)))

    env.execute(" window test")


  }

}

